package zhongche;

import com.alibaba.fastjson.JSONObject;
import com.zhongche.bean.TrainInfo;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;
import java.util.Set;

public class FlinkCepCheckDemo {

    private static final String SPEED_KEY = "s0915691";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String jsonStr0 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 0,\n" +
                " \"s1999915\": 1\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716155040\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        String jsonStr1 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 0,\n" +
                " \"s1999915\": 1\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716155527\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716155040\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";
        String jsonStr2 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 60,\n" +
                "    \"s1999915\": 1\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716173626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        KeyedStream<TrainInfo, Integer> infoIntegerKeyedStream = env.fromElements(jsonStr0,jsonStr1, jsonStr2)
                .map(new MapFunction<String, TrainInfo>() {
                    @Override
                    public TrainInfo map(String value) throws Exception {
                        return JSONObject.parseObject(value, TrainInfo.class);
                    }
                })
                .name("转换对象")
                .keyBy(TrainInfo::getTrainId);

        Pattern<TrainInfo, TrainInfo> pattern = Pattern.<TrainInfo>begin("start")
                .where(new SimpleCondition<TrainInfo>() {
                    @Override
                    public boolean filter(TrainInfo value) throws Exception {
                        System.out.println(value);
                        return value.getDataParam().get(SPEED_KEY) == 0;
                    }
                })
                .next("model")
                .where(new SimpleCondition<TrainInfo>() {
                    @Override
                    public boolean filter(TrainInfo value) throws Exception {
                        return value.getDataParam().get(SPEED_KEY) == 0;
                    }
                })
                .oneOrMore()
                .optional()
                .until(new SimpleCondition<TrainInfo>() {
                    @Override
                    public boolean filter(TrainInfo value) throws Exception {
                        return value.getDataParam().get(SPEED_KEY) > 0;
                    }
                });

        PatternStream<TrainInfo> trainInfoPatternStream = CEP.pattern(infoIntegerKeyedStream, pattern);


        SingleOutputStreamOperator<String> process = trainInfoPatternStream.process(new PatternProcessFunction<TrainInfo, String>() {
            @Override
            public void processMatch(Map<String, List<TrainInfo>> match, Context ctx, Collector<String> out) throws Exception {
                Set<Map.Entry<String, List<TrainInfo>>> entries = match.entrySet();
                for (Map.Entry<String,List<TrainInfo>> entry: entries) {
                    System.out.println(entry.getKey());
                    entry.getValue().forEach(value ->{
                        System.out.println(JSONObject.toJSONString(value));
                    });
                }
            }
        });
        process.print("cep");
        env.execute();
    }


    public static class MySource implements SourceFunction<JSONObject> {

        private Boolean isRuning = true;

        private JSONObject jsonObject;

        private Integer speed;

        private Long startDateTime;

        public MySource(Integer speed, Long startDateTime) {
            this.speed = speed;
            this.startDateTime = startDateTime;
            this.jsonObject = JSONObject.parseObject("{\n" +
                    "  \"box_num\": 1,\n" +
                    "  \"current_station\": 5,\n" +
                    "  \"dDevice\": 50,\n" +
                    "  \"dataParam\": {\n" +
                    "\"s0000108\": 0,\n" +
                    "\"s0915691\": " + this.speed + ",\n" +
                    "    \"s1999915\": 1\n" +
                    "  },\n" +
                    "  \"data_sDeviceNo\": 20,\n" +
                    "  \"frame_num\": 5657,\n" +
                    "  \"line_id\": 2,\n" +
                    "  \"line_num\": \"2\",\n" +
                    "  \"message_type\": 5025,\n" +
                    "  \"netLink_flag\": 0,\n" +
                    "  \"next_station\": 0,\n" +
                    "  \"ocs_time\": \"210716153627\",\n" +
                    "  \"sDevice\": 40,\n" +
                    "  \"sDeviceTime\": \"" + this.startDateTime + "\",   \n" +
                    "  \"second_circle\": 1,\n" +
                    "  \"speed\": 0.0,\n" +
                    "  \"start_station\": 5,\n" +
                    "  \"terminal_station\": 28,\n" +
                    "  \"train_id\": 159160,\n" +
                    "  \"train_name\": \"159160\",\n" +
                    "  \"train_num\": 159160,\n" +
                    "  \"train_type\": \"A52\",\n" +
                    "  \"ttype_id\": 10\n" +
                    "}\n");
        }

        @Override
        public void run(SourceContext<JSONObject> ctx) throws Exception {
            while (isRuning) {
                ctx.collect(this.jsonObject);
                Thread.sleep(1000L);
                this.speed -= 5;
                if(this.speed <= 0){
                    this.speed = 0;
                }
            }
        }

        @Override
        public void cancel() {
            isRuning = false;
        }
    }

}
